package com.enlorenz.core.channel.activemq;

import java.io.IOException;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import com.enlorenz.core.channel.IChannelEngine;
import com.enlorenz.core.engine.exception.EngineControlException;
import com.enlorenz.core.engine.imp.thread.quick.AbstractQuickControlThread;
import com.enlorenz.core.queue.transfer.imp.WaitMessageQueue;

public class MQServerChannel extends AbstractQuickControlThread implements IChannelEngine{
	 // ConnectionFactory ：连接工厂，JMS 用它创建连接
    private ConnectionFactory connectionFactory;
    // Connection ：JMS 客户端到JMS Provider 的连接
    private Connection connection = null;
    // Session： 一个发送或接收消息的线程
    private Session session;
    // Destination ：消息的目的地;消息发送给谁.
    private Destination destination;
    // 消费者，消息接收者
    private MessageConsumer consumer;
    //MQ服务地址
	private String mqUrl;
	//创建通道类型
	private String mqName;
	//发送是否成功
	private boolean sendFlag=true;
	
	private WaitMessageQueue recvQueue;
    
    public void setRecvQueue(WaitMessageQueue recvQueue) {
		this.recvQueue = recvQueue;
	}

	/**
	 * 启动MQ客户端
	 */
	protected void startEngine() throws EngineControlException {
		try {
			//启动服务
			serverStart();
			super.startEngine();
		} catch (Exception e) {
			e.printStackTrace();
			close();
			new EngineControlException(this.getName(),e);
		}
	}
	
	/**
	 * Socket是否还存活
	 */
	@Override
	public boolean isAlive() {
		return sendFlag && super.isAlive();
	}
	
	/**
	 * MQ启动
	 * 
	 * @throws IOException
	 */
	private void serverStart() throws Exception {
		connectionFactory = new ActiveMQConnectionFactory(
				ActiveMQConnection.DEFAULT_USER,
				ActiveMQConnection.DEFAULT_PASSWORD, mqUrl);
		// 构造从工厂得到连接对象
		connection = connectionFactory.createConnection();
		// 启动
		connection.start();
		// 获取操作连接
		session = connection.createSession(Boolean.FALSE,
				Session.AUTO_ACKNOWLEDGE);
		// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
		destination = session.createQueue(mqName);
		consumer = session.createConsumer(destination);
	}
    
	/**
	 * 关闭打开的链接
	 */
	private void close() {
		if(null!=consumer){
			try {
				consumer.close();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
		
		if(null!=session){
			try {
				session.close();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
		if(null!=connection){
			try {
				connection.stop();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
		if(null!=connection){
			try {
				connection.close();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
	}
	
	@Override
	protected void handle() throws Exception {
		try{
	        //设置接收者接收消息的时间，为了便于测试，这里谁定为100s
			ObjectMessage message = (ObjectMessage)consumer.receive(1000);
			//System.out.println(this.getMqName()+":"+message);
			if (null != message) {
				recvQueue.addMsg(message.getObject());
	        } 
			sendFlag=true;
		}catch (Exception e) {
			e.printStackTrace();
			sendFlag=false;
			Thread.sleep(5000);
		}
	}
	
	

	public String getMqUrl() {
		return mqUrl;
	}

	public void setMqUrl(String mqUrl) {
		this.mqUrl = mqUrl;
	}

	public String getMqName() {
		return mqName;
	}

	public void setMqName(String mqName) {
		this.mqName = mqName;
	}
}
